/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.util;

import cn.ac.iie.di.ban.ycl.wx.nrd.update.exception.UpdateException;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.QueryOperators;
import com.mongodb.ServerAddress;
import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;
import org.bson.Document;

/**
 *
 * @author zwlin
 */
public class MongoHelper {

    private static final Logger logger = Logger.getLogger(MongoHelper.class.getName());

    MongoClient conn;
    MongoDatabase db;
    public MongoCollection<Document> collection;

    String[] retFieldNames;

    public MongoHelper(String[] hosts, String dbname, String tablename, String username, String passwd) throws UpdateException {
        try {
            List<ServerAddress> seeds = new ArrayList<>();
            for (String host : hosts) {
                String[] serv = host.split(":");
                if (serv != null && serv.length == 2) {
                    seeds.add(new ServerAddress(serv[0], Integer.parseInt(serv[1])));
                }
            }
            MongoCredential credential = MongoCredential.createScramSha1Credential(username, dbname, passwd.toCharArray());
            List<MongoCredential> credentialsList = new ArrayList<MongoCredential>();
            credentialsList.add(credential);

            conn = new MongoClient(seeds, credentialsList);
            db = conn.getDatabase(dbname);
            collection = db.getCollection(tablename);
        } catch (Exception ex) {
            try {
                conn.close();
            } catch (Exception ex1) {
            }
            throw new UpdateException("cacnnot init mongo helper !", ex);

        }
    }

    public static Document getDocument(HashMap<String, Object> rec) {
        Document doc = new Document();
        for (Map.Entry<String, Object> entry : rec.entrySet()) {
            Object v = entry.getValue();
            if (v.getClass().isArray()) {
                int n = Array.getLength(v);
                List l = new ArrayList(n);
                for (int i = 0; i < n; ++i) {
                    l.add(Array.get(v, i));
                }
                doc.put(entry.getKey(), l);
            } else {
                doc.put(entry.getKey(), v);
            }
        }
        return doc;
    }

    /**
     * 批量查询
     *
     * @param keyCol	用来定位记录的列名
     * @param keys	用来定位记录的列值集合
     * @param retFields	返回的列名
     * @return
     */
    public List<HashMap<String, Object>> find(String keyCol, Collection<String> keys, String[] retFields) {
        try {
            List<HashMap<String, Object>> rtn = new ArrayList<>(keys.size());
            Document projection = new Document();
            for (String str : retFields) {
                projection.put(str, 1);
            }
            MongoCursor<Document> cursor = collection.find(new Document(keyCol, new BasicDBObject(QueryOperators.IN, keys))).projection(projection).iterator();
            while (cursor.hasNext()) {
                Document rec = cursor.next();
                HashMap<String, Object> rtnRec = new HashMap<>(retFields.length);
                for (String str : retFields) {
                    rtnRec.put(str, rec.get(str));
                }
                rtn.add(rtnRec);
            }
            return rtn;
        } catch (Exception e) {
            logger.error("find error", e);
            return null;
        }

    }

    /**
     * 批量更新，将某一列的值增加一
     *
     * @param keyCol	用于定位记录的列名
     * @param keys	用于定位记录的列值集合
     * @param incCol	需要加一的列名
     */
    public void upinc(String keyCol, Collection<String> keys, String incCol) {
        List<WriteModel<Document>> updatedoc = new ArrayList<>();
        for (String key : keys) {
            Document doc = new Document();
            doc.put(incCol, 1L);
            updatedoc.add(
                    new UpdateOneModel<Document>(new Document(keyCol, key),
                            new Document("$inc", doc), new UpdateOptions().upsert(false)));
        }
        collection.bulkWrite(updatedoc);
    }

    /**
     * 批量更新，将某一列的值增加一定数值。
     *
     * @param keyCol	用于定位记录的列名
     * @param updateCol	用于增加的列名
     * @param updateValues	keyCol的值和updateCol的增量的对应关系
     */
    public void upadd(String keyCol, String updateCol, HashMap<String, Long> updateValues) {
        List<WriteModel<Document>> updatedoc = new ArrayList<>();
        for (Map.Entry<String, Long> entry : updateValues.entrySet()) {
            Document doc = new Document();
            doc.put(updateCol, entry.getValue());
            updatedoc.add(
                    new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
                            new Document("$inc", doc), new UpdateOptions().upsert(false)));
        }
        collection.bulkWrite(updatedoc);
    }

    /**
     * 批量更新，将某一列的值更新为新值。
     *
     * @param keyCol	用于定位记录的列名
     * @param updateCol	用于更新的列名
     * @param updateValues	keyCol的值和updateCol的增量的对应关系
     */
    public void update(String keyCol, String updateCol, HashMap<String, Object> updateValues) {
        List<WriteModel<Document>> updatedoc = new ArrayList<>();
        for (Map.Entry<String, Object> entry : updateValues.entrySet()) {
            Document doc = new Document();
            doc.put(updateCol, entry.getValue());
            updatedoc.add(
                    new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
                            new Document("$set", doc), new UpdateOptions().upsert(false)));
        }
        collection.bulkWrite(updatedoc);
    }

    public void update(String keyCol, HashMap<String/*update_keycol_value*/, HashMap<String, Object>> updateMap) {
        try {
            List<WriteModel<Document>> updatedoc = new ArrayList<>();
            for (Map.Entry<String, HashMap<String, Object>> entry : updateMap.entrySet()) {
//                    logger.info(entry.getKey()+"---"+entry.getValue());
                Document doc = new Document();
                doc.putAll(entry.getValue());
                updatedoc.add(
                        new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
                                new Document("$set", doc), new UpdateOptions().upsert(false)));
            }
            BulkWriteResult bulkWrite = collection.bulkWrite(updatedoc);
//                logger.info("update to mpngodb "+bulkWrite.getMatchedCount());
            logger.info("update to mpngo " + bulkWrite.getModifiedCount());
        } catch (Exception e) {
            logger.info("this is null");
        }

    }

}
